package com.xiaoran.RMQ_1.test4;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.xiaoran.RMQ_1.utlis.RabbitMqUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class ConfirmMessage_Produce {
    public static final  int number=1000;
    public static void main(String args[]) throws Exception{
        //单个确认 消耗的时间是：1164ms
        //ConfirmMessage_Produce.publicshMessageIndividually();
        //批量确认  消耗的时间是：127ms
        //ConfirmMessage_Produce.publicshMessageBatch();
        //异步批量确认 消耗的时间是：55ms
        ConfirmMessage_Produce.publishMessageAsync();
    }
    //单个确认
    public static void publicshMessageIndividually() throws Exception{
        Channel channel=RabbitMqUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //创建队列
        String name=UUID.randomUUID().toString().substring(0,2);
        channel.queueDeclare(name,true,false,false,null);
        //发送消息
        long begin=System.currentTimeMillis();
        for (int i=0;i<number;i++){
            String message=i+" ";
            channel.basicPublish("",name,null,message.getBytes());
            //单个消息发布之后就会确认
            boolean flag=channel.waitForConfirms();
            if(flag){
                System.out.println("消息发布成功。。。。。。。。。。。。");
            }
        }
        long  end=System.currentTimeMillis();
        System.out.println("消耗的时间是："+(end-begin)+"ms");
    }

    //批量确认
    public static void publicshMessageBatch() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //创建队列
        String name = UUID.randomUUID().toString().substring(0, 2);
        channel.queueDeclare(name, true, false, false, null);
        //发送消息
        long begin = System.currentTimeMillis();
        int aa=100;
        for (int i=0;i<number;i++){
            String message=i+" ";
            channel.basicPublish("",name,null,message.getBytes());
            if (i%aa==0){
                boolean cc=channel.waitForConfirms();
                if(cc){
                    System.out.println("消息确认成功");
                }
            }
        }
        long  end=System.currentTimeMillis();
        System.out.println("消耗的时间是："+(end-begin)+"ms");
    }

    //异步确认
    public  static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //创建一个用于存储消息失败的map
        ConcurrentSkipListMap<Long,String> concurrentSkipListMap=new ConcurrentSkipListMap<>();
        //创建队列
        String name = UUID.randomUUID().toString().substring(0, 2);
        channel.queueDeclare(name, true, false, false, null);
        //成功的消息监听
        ConfirmCallback ackCallback=(deliveryTag,multiple)->{
            //2.去除掉成功的消息
            //2.1 判断是否为批量删除
            if (multiple){
              concurrentSkipListMap.headMap(deliveryTag).clear();
            }else{
                concurrentSkipListMap.remove(deliveryTag);
            }
          System.out.println("成功的消息"+deliveryTag);
        };
        //失败的消息监听器
        //deliveryTag  消息的标记
        // multiple   是否批量确认
        ConfirmCallback nackCallback=(deliveryTag,multiple)->{
            String message=concurrentSkipListMap.get(deliveryTag);
            System.out.println("***************************************");
            System.out.println("未确认的消息是谁"+message+"失败的消息："+deliveryTag);
        };
        //准备消息的监听器，监听器告诉MQ哪些消息成功哪些失败
        channel.addConfirmListener(ackCallback,nackCallback);
        //发送消息
        long begin = System.currentTimeMillis();
        for (int i=0;i<number;i++){
            String message=""+i;
            channel.basicPublish("",name,null,message.getBytes());
            //1..记录下所有要发送的消息总和
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),message);
        }
        long  end=System.currentTimeMillis();
        System.out.println("消耗的时间是："+(end-begin)+"ms");

    }
}
